package nl.altenpts.net.broker;

import java.nio.ByteBuffer;
import java.security.PublicKey;
import java.security.SignatureException;
import java.util.HashMap;

import nl.altenpts.net.NetReactor;


/**
 * Request broker
 * 
 * @author fabian
 *
 */
public class RequestBroker {
	
	
	/**
	 * Construct a new request broker.
	 * 
	 * @param broker The signed message broker with which messages will be sent
	 * @param reactor NetReactor
	 */
	public RequestBroker(SignedMessageBroker broker, NetReactor reactor) {
		this.msgbroker = broker;
		this.reactor = reactor;
		SignedMessageHandler handler = new SignedMessageHandler() {

			public void onIncoming(IncomingMessage message, PublicKey key) {
				
				byte b  = message.getPayload().get();
				long id = message.getPayload().getLong();
				
				
				switch(b) {
				case REQUESTBROKER_REQUEST:					
					if(reqhandler != null) {
						IncomingRequest inrequest = new IncomingRequest(message, id, msgbroker);
						reqhandler.onIncoming(inrequest, key);
					}

					break;
				case REQUESTBROKER_RESPONSE:
					ResponseHandler handler;
					synchronized(pending) {
						handler = pending.remove(id);
					}
					if(handler != null) {
						handler.onResponse(message);
					}
					break;					
				}
			}
			
		};
		this.msgbroker.setHandler(handler);
	}
	
	
	/**
	 * Sends a new request. A response or timeout is passed on to handler.
	 * 
	 * @param message request message
	 * @param handler response handler
	 * @param timeout timeout in milliseconds
	 * @throws SignatureException
	 */
	public void sendRequest(OutgoingMessage message, ResponseHandler handler, int timeout) {
		
		long thisrequestid;

		synchronized(pending) {
			thisrequestid = requestid;
			{
				if(requestid != Long.MAX_VALUE) {
					requestid++;
				} else {
					requestid = Long.MIN_VALUE;				
				}
			} while(pending.containsKey(requestid));
		}
		
		ByteBuffer buffer = ByteBuffer.allocate(1+8);
		
		buffer.put(REQUESTBROKER_REQUEST);
		buffer.putLong(thisrequestid);
		buffer.flip();
		message.prependContent(buffer);
		
		synchronized(pending) {
			pending.put(thisrequestid, handler);
		}
		msgbroker.sendMessage(message);
		reactor.addTask(new TimeoutHandler(thisrequestid), System.currentTimeMillis() + timeout , false);
				
	}
	
	/**
	 * Sets a new handler
	 * 
	 * @param handler new handler
	 */
	public void setHandler(RequestHandler handler) {
		reqhandler = handler;
	}
	
	/**
	 * Returns the current request handler
	 * 
	 * @return current request handler
	 */
	public RequestHandler getHandler() {
		return reqhandler;
	}
	/**
	 * 
	 */
	
	/**
	 *  Returns the signed message broker used by
	 *  this request broker
	 *  
	 *  @return Underlying request broker
	 */
	public SignedMessageBroker getBroker() {
		return msgbroker;
	}
	
	
	private RequestHandler reqhandler = null;
	private SignedMessageBroker msgbroker;
	private NetReactor reactor;
	private HashMap<Long, ResponseHandler> pending = new HashMap<Long, ResponseHandler>();	
	private long requestid = Long.MIN_VALUE + 1;
	
	final static byte REQUESTBROKER_REQUEST  = 1;
	final static byte REQUESTBROKER_RESPONSE = 2;
	
	
	/**
	 * Handler for timeout tasks.
	 * 
	 * @author fabian
	 *
	 */
	class TimeoutHandler implements Runnable {
		public TimeoutHandler(long id) {
			this.id = id;
		}

		public void run() {
			synchronized(pending) {
				ResponseHandler handler = pending.remove(id);
				if(handler != null) {
					System.out.println("Message Timed out");
					handler.onTimeout();
				}
			}
		}		
		private long id;		
	}
}
